package com.mini.grpc.client.netty;

import com.mini.grpc.client.CallOptions;
import com.mini.grpc.client.ClientCall;
import com.mini.grpc.client.ManagedChannel;
import com.mini.grpc.common.MethodDescriptor;
import com.mini.grpc.common.Status;
import com.mini.grpc.transport.ClientTransport;
import com.mini.grpc.transport.TransportListener;
import com.mini.grpc.transport.netty.Http2Transport;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * NettyChannel是基于Netty的通道实现。
 */
public class NettyChannel extends ManagedChannel {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyChannel.class);
    
    private final SocketAddress address;
    private final ClientTransport transport;
    private final AtomicBoolean shutdown = new AtomicBoolean(false);
    private final AtomicBoolean terminated = new AtomicBoolean(false);
    private final CountDownLatch terminationLatch = new CountDownLatch(1);
    private final TransportListener transportListener;
    
    /**
     * 创建一个新的NettyChannel
     *
     * @param address 服务端地址
     * @param useSSL 是否使用SSL
     */
    public NettyChannel(SocketAddress address, boolean useSSL) {
        this.address = address;
        this.transportListener = new ChannelTransportListener();
        this.transport = new Http2Transport(useSSL).createClient(address, transportListener);
        
        // 启动传输
        if (!transport.start()) {
            throw new RuntimeException("Failed to start transport");
        }
    }
    
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> method, CallOptions options) {
        System.out.println("=== NettyChannel.newCall called for method: " + method.getFullMethodName() + " ===");
        System.out.println("Transport: " + transport.getClass().getName());
        
        NettyClientCall<ReqT, RespT> call = new NettyClientCall<>(method, options, transport);
        System.out.println("Created NettyClientCall: " + call);
        
        return call;
    }
    
    @Override
    public ConnectivityState getState(boolean requestConnection) {
        if (terminated.get()) {
            return ConnectivityState.SHUTDOWN;
        }
        
        if (shutdown.get()) {
            return ConnectivityState.SHUTDOWN;
        }
        
        // 将传输层状态映射到通道状态
        switch (transport.getState()) {
            case IDLE:
                return ConnectivityState.IDLE;
            case CONNECTING:
                return ConnectivityState.CONNECTING;
            case READY:
                return ConnectivityState.READY;
            case TRANSIENT_FAILURE:
                return ConnectivityState.TRANSIENT_FAILURE;
            case SHUTDOWN:
                return ConnectivityState.SHUTDOWN;
            default:
                return ConnectivityState.IDLE;
        }
    }
    
    @Override
    public ManagedChannel shutdown() {
        if (shutdown.compareAndSet(false, true)) {
            logger.info("Shutting down channel to {}", address);
            transport.shutdown(Status.OK.withDescription("Channel shutdown requested"));
        }
        return this;
    }
    
    @Override
    public ManagedChannel shutdownNow() {
        if (shutdown.compareAndSet(false, true)) {
            logger.info("Shutting down channel to {} now", address);
            transport.shutdown(Status.CANCELLED.withDescription("Channel shutdown now requested"));
        }
        return this;
    }
    
    @Override
    public boolean isShutdown() {
        return shutdown.get();
    }
    
    @Override
    public boolean isTerminated() {
        return terminated.get();
    }
    
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return terminationLatch.await(timeout, unit);
    }
    
    @Override
    public void awaitTermination() throws InterruptedException {
        terminationLatch.await();
    }
    
    /**
     * 传输监听器，用于接收传输层事件
     */
    private class ChannelTransportListener implements TransportListener {
        
        @Override
        public void transportReady() {
            logger.info("Transport ready to {}", address);
        }
        
        @Override
        public void transportInUse(Status status) {
            logger.warn("Transport to {} in use: {}", address, status);
        }
        
        @Override
        public void transportTerminated(Status status) {
            logger.info("Transport to {} terminated: {}", address, status);
            terminated.set(true);
            terminationLatch.countDown();
        }
    }
} 